package com.yanggu.bigdata.realtime.app.function

import cn.hutool.core.util.StrUtil
import cn.hutool.json.JSONUtil
import com.alibaba.fastjson.{JSON, JSONObject}
import com.yanggu.bigdata.realtime.app.BroadcastObject.tableProcessMapStateDescriptor
import com.yanggu.bigdata.realtime.app.OutputTagObject.dimHbaseOutputTag
import com.yanggu.bigdata.realtime.bean.TableProcess
import com.yanggu.bigdata.realtime.common.GmallConfig
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction
import org.apache.flink.util.Collector

import java.sql.{Connection, DriverManager, PreparedStatement}
import java.util.{Properties, StringJoiner}

/**
 *
 * @tparam IN1 String 主流数据
 * @tparam IN2 String 广播流数据
 * @tparam OUT JsonObject 主流输出数据
 *
 */
class TableProcessBroadcastProcessFunction extends BroadcastProcessFunction[String, String, JSONObject] {

  //初始化链接, 同时使用懒加载的方式
  private lazy val connection: Connection = {
    Class.forName(GmallConfig.PHOENIX_DRIVER)
    val properties = new Properties
    //phoenix查询时需要设置这个参数
    properties.setProperty("phoenix.schema.isNamespaceMappingEnabled", "true")
    val tempConnection = DriverManager.getConnection(GmallConfig.PHOENIX_SERVER, properties)
    //设置库名
    tempConnection.setSchema(GmallConfig.HBASE_SCHEMA)
    tempConnection
  }

  private var preparedStatement: PreparedStatement = _

  //处理主流数据
  //value的数据格式
  /*
  {
       "database":"",
       "table":"",
       //删除或者修改之前的数据
      "before":"json字符串",
       //新增或者修改后的数据
       "after":"json字符串",
       "operation":"操作类型:insert、update、delete"
   }
   */
  override def processElement(value: String,
                              ctx: BroadcastProcessFunction[String, String, JSONObject]#ReadOnlyContext,
                              out: Collector[JSONObject]): Unit = {
    val data = JSON.parseObject(value)
    //获取表名
    val table = data.getString("table")
    //获取操作类型
    val operation = data.getString("operation")

    //通过table和operation获取tableProcess
    val tableProcess = ctx.getBroadcastState(tableProcessMapStateDescriptor).get((table, operation))
    //如果为空, 直接return结束
    if (tableProcess == null) {
      println(s"表名: $table, 操作类型: ${operation}不存在")
      return
    }

    //根据配置的字段进行过滤不需要的字段
    val columns = tableProcess.sinkColumns.split(",")
    //获取数据
    val after = JSON.parseObject(data.getString("after"))
    //删除不需要的字段
    after.entrySet().removeIf(entry => !columns.contains(entry.getKey))
    //把过滤字段后的数据重新保存
    data.put("after", JSONUtil.toJsonStr(after))

    //将hbase表名或者kafka的dwd的topic放入数据中
    data.put("sinkTable", tableProcess.sinkTable)

    //根据tableProcess的配置进行分流(维度表数据到Hbase、事实表数据到Kafka)
    tableProcess.sinkType match {
      case "hbase" =>
        ctx.output(dimHbaseOutputTag, data)
      case "kafka" =>
        //kafka放入主流中
        out.collect(data)
    }
  }

  //处理广播流数据, 广播流中的数据是配置表的变化数据
  //value的数据格式
  /*
    {
      "database": "gmall_flink",
      //删除或者修改之前的数据
      "before":"json字符串",
      //新增或者修改后的数据
      "after": "{\"category_level\":3,\"id\":95,\"category_id\":997,\"attr_name\":\"价格\"}",
      //操作类型:insert、update、delete、read
      "operation": "read",
      "table": "base_attr_info"
    }
   */
  override def processBroadcastElement(value: String,
                                       ctx: BroadcastProcessFunction[String, String, JSONObject]#Context,
                                       out: Collector[JSONObject]): Unit = {
    //1. 转换成JsonObject对象
    val jsonObject = JSON.parseObject(value)

    //获取操作类型
    val operation = jsonObject.getString("operation")

    //获取广播状态
    val broadcastState = ctx.getBroadcastState(tableProcessMapStateDescriptor)

    var after = jsonObject.getString("after")
    after = if (StrUtil.isEmptyIfStr(after)) jsonObject.getString("before") else after
    val data = JSON.parseObject(after, classOf[TableProcess])

    //如果输出到hbase, 检查表是否创建, 是否需要添加字段
    if (data.sinkType == "hbase") {
      checkSchema(data)
    }

    //根据操作类型进行判断
    operation match {
      //如果是新增、修改、读操作
      case "insert" | "update" | "read" =>
        //写入到状态中, 广播出去
        broadcastState.put((data.sourceTable, data.operateType), data)
      //如果是删除数据, 删除广播变量中的数据
      case "delete" =>
        broadcastState.remove((data.sourceTable, data.operateType))
    }

  }

  //建表语句 : create table if not exists db.tn(id varchar primary key,tm_name varchar) xxx;
  //查看表是否创建, 或者修改表的字段
  //这里只实现了创建表的逻辑, 关于修改表或者是删除表的逻辑未实现
  def checkSchema(data: TableProcess): Unit = {
    val joiner = new StringJoiner(", ", "(", ")")
    val pk = if (StrUtil.isEmptyIfStr(data.sinkPk)) "id" else data.sinkPk
    data.sinkColumns.split(",")
      .foreach(column => {
        var sql = ""
        //判断是否为主键
        if (column == pk) {
          sql = column + " varchar PRIMARY KEY"
        } else {
          sql = column + " varchar"
        }
        joiner.add(sql)
      })
    val sql = s"CREATE TABLE IF NOT EXISTS ${GmallConfig.HBASE_SCHEMA}.${data.sinkTable} ${joiner.toString}"
    println(sql)
    //预编译sql
    preparedStatement = connection.prepareStatement(sql)
    //执行sql
    preparedStatement.execute()
  }

}
